package MR.NBA;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class NBAChampionCount {
	public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
		Text newKey = new Text();
		IntWritable newValue = new IntWritable(1);  // 每一行数据都是冠军，所以直接赋值为1
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
			String[] split = value.toString().split(",");
			String Champion = split[2];
			// 获取分区标识
			String area = null ;
			if (split.length == 6) 
				area = split[5];
			// 如果有分区，则与冠军球队一起写入key中，并以“，”分割，没有分区，就直接输出冠军队名
			if (area != null)
				newKey.set(Champion + "," + area);
			else
				newKey.set(Champion);
			context.write(newKey,newValue);
		}

	}
	
	
	public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		IntWritable iw = new IntWritable();
		@Override
		protected void reduce(Text key, Iterable<IntWritable> value, Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable iw : value) {
				sum += iw.get(); 
			}
			iw.set(sum);
			context.write(key, iw);
		}
	}
	
	public static class MyPartioner extends Partitioner<Text, IntWritable> {
		@Override
		public int getPartition(Text key,IntWritable value, int numParttitions) {
			String[] split = key.toString().split(",");
			String area = null;
			if (split.length > 1) {
				area = split[1];
			}
			
			if ("E".equals(area)) {
				return 0;
			} else if ("W".equals(area)) {
				return 1;
			} else {
				return 2;
			}
		}
}
	
	// 主方法
	public static void main(String[] args) throws Exception{
		Configuration conf = new Configuration();
		
		Job job = Job.getInstance(conf);
		job.setJarByClass(NBAChampionCount.class);
		
		job.setMapperClass(MyMapper.class);
		job.setReducerClass(MyReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		job.setPartitionerClass(MyPartioner.class);
		
		job.setNumReduceTasks(3);
		
		String input = "hdfs://namenode:8020/NBAMRoutput/part-m-00000";
		String output = "hdfs://namenode:8020/NBAMRoutput/ChampionCount";
		
		FileInputFormat.setInputPaths(job,new Path(input));
		FileOutputFormat.setOutputPath(job,new Path(output));
		
		if(job.waitForCompletion(true)) {
			System.out.println("ִ运行成功!");
			System.exit(0);
		}else {
			System.out.println("ִ运行失败...");
			System.exit(1);
		}
	}	
}

